package com.example.springboottest.example.completablefuture;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiConsumer;

import static java.util.stream.Collectors.toList;

/**
 * CompletableFuture详细请查看
 * https://www.lizenghai.com/archives/42010.html
 */
public class ClientTest {
    static List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("1Shop"),
            new Shop("2Shop"),
            new Shop("3Shop"),
            new Shop("4Shop"),
            new Shop("5Shop"),
            new Shop("6Shop"),
            new Shop("7Shop"),
            new Shop("8Shop"),
            new Shop("9Shop"),
            new Shop("10Shop"),
            new Shop("11Shop"),
            new Shop("12Shop"),
            new Shop("13Shop"),
            new Shop("14Shop"),
            new Shop("15Shop"),
            new Shop("16Shop"),
            new Shop("17Shop"));

    static final Executor executor =
            Executors.newFixedThreadPool(Math.min(shops.size(), 100),
                    new ThreadFactory() {
                        public Thread newThread(Runnable r) {
                            Thread t = new Thread(r);
                            t.setDaemon(true);
                            return t;
                        }
                    });

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        findPrice();

//        long start = System.nanoTime();
//        System.out.println(findPrices("myPhone27S"));
//        long duration = (System.nanoTime() - start) / 1_000_000;
//        System.out.println("Done in " + duration + " msecs");

        allOfTest();
    }

    public static void findPrice() {
        //模拟客户端查询价格
        Shop shop = new Shop("BestShop");
        long start = System.nanoTime();
        Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
        long invocationTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("Invocation returned after " + invocationTime
                + " msecs");
        // 执行更多任务，比如查询其他商店
        //        doSomethingElse();
        // 在计算商品价格的同时
        try {
            double price = futurePrice.get();
            System.out.printf("Price is %.2f%n", price);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("Price returned after " + retrievalTime + " msecs");
    }

    /**
     * 本机器是8Cores16Threads
     * <p>
     * 方案1肯定不行
     * 方案2和方案3再不修改默认线程池线程数量的情况下，不分伯仲
     * 默认线程池线程数量等于cpu的核数 也就是Runtime.getRuntime().availableProcessors()
     * 但方案3可以自定义线程池核心数，也就是方案4
     *
     * @param product
     * @return
     */
    public static List<String> findPrices(String product) {
        System.out.println("线程池默认线程数:\t" + Runtime.getRuntime().availableProcessors());
        /* 方案1 最慢的同步堵塞执行 17个商店  大约17s+
        return shops.stream()
                .map(shop -> String.format("%s price is %.2f",
                        shop.getName(), shop.getPrice(product)))
                .collect(toList());
        */

        /* 方案2 较快的并行流执行 17个商店  大约2s
        return shops.parallelStream()
                .map(shop -> String.format("%s price is %.2f",
                        shop.getName(), shop.getPrice(product)))
                .collect(toList());
        */

        /* 方案3 future方式 17个商店  大约2s
        List<CompletableFuture<String>> priceFutures =
                shops.stream()
                        .map(shop -> CompletableFuture.supplyAsync(
                                () -> shop.getName() + " price is " +
                                        shop.getPrice(product)))
                        .collect(toList());
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(toList());
        */

        /* 方案4 使用自定义线线程池 17个商店  大约1s*/
        List<CompletableFuture<String>> priceFutures =
                shops.stream()
                        .map(shop -> CompletableFuture.supplyAsync(
                                () -> shop.getName() + " price is " +
                                        shop.getPrice(product), executor))
                        .collect(toList());
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(toList());
    }

    private static void allOfTest() throws ExecutionException, InterruptedException {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(8);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "f1";
        }).whenCompleteAsync(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) {
                System.out.println(System.currentTimeMillis() + ":" + s);
            }
        });

        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "f2";
        }).whenCompleteAsync(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) {
                System.out.println(System.currentTimeMillis() + ":" + s);
            }
        });

        CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "f3";
        }).whenCompleteAsync(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) {
                System.out.println(System.currentTimeMillis() + ":" + s);
            }
        });

        /**
         * applyToEither
         * CompletableFuture最快的执行完成的结果作为下一个stage的输入结果
         * acceptEither
         * 最快CompletableFuture执行完成的时候，action消费就会得到执行。
         * runAfterEither
         * 任何一个CompletableFuture完成之后，就会执行下一步的Runnable。
         * anyOf
         * 任何一个CompletableFuture完成，anyOf函数就会返回。
         * allof
         * 线程阻塞，等待全部执行完成
         */
        CompletableFuture<Object> all = CompletableFuture.anyOf(f1, f2, f3);

        //阻塞，直到所有任务结束。
        System.out.println(System.currentTimeMillis() + ":阻塞");
        all.join();
        System.out.println(System.currentTimeMillis() + ":阻塞结束");

        //一个需要耗时2秒，一个需要耗时3秒，只有当最长的耗时3秒的完成后，才会结束。
        System.out.println("任务均已完成。");
    }


}
